package com.atguigu.chapter07;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.Properties;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/3 14:19
 */
public class Flink19_Watermark_Summary {
    public static void main(String[] args) throws Exception {
        /**
         *
         * TODO 1、watermark概念
         *      1） 衡量 事件时间 的进展
         *      2） 单调不减的（保持不变，或 增加）
         *      3） 是一个特殊时间戳，生成之后插入到流里，随着流的流动 传递
         *      4） 解决 乱序 的问题
         *      5） 认为，在它之前的数据都处理过了（如果还有，说明该数据迟到了）
         *      6） 触发 窗口等 的 计算、关闭
         *
         * TODO 2、watermark 写法
         *      1)升序写法
         *      WatermarkStrategy
         *            .<T>forMonotonousTimestamps()
         *            .withTimestampAssigner()
         *      2）乱序写法
         *      WatermarkStrategy
         *            .<T>forBoundedOutOfOrderness(Duration)
         *            .withTimestampAssigner()
         *
         * TODO 3、watermark 分类
         *      1）间歇性的生成： 来一条数据，更新一次   => onEvent()
         *      2）周期性的生成（默认）： 间隔固定周期，更新一次，默认200ms   => onPeriodicEmit()
         *
         * TODO 4、watermark 生成逻辑
         *      1) 升序
         *          watermark = 当前最大事件时间 - 1ms
         *      2）乱序
         *          watermark = 当前最大事件时间 - 乱序程度 - 1ms
         *
         * TODO 5、watermark 在多并行度下的传递
         *      1） 一对多 ： 广播
         *      2） 多对一 ： 取最小
         *
         * TODO 6、Flink处理乱序和迟到
         *      1） watermark解决乱序
         *      2）窗口允许迟到，解决关窗前的迟到数据
         *           当 watermark >= end - 1 时， 触发窗口计算
         *           当 end - 1ms < watermark < end + 允许迟到时间 -1 ms 时，每来一条迟到数据，都会触发一次计算
         *           当 watermark >= end + 允许迟到时间 -1 ms, 就会 关闭窗口，再有迟到的数据来，也不处理了
         *      3）侧输出流，解决关窗之后的迟到数据
         *
         * TODO 7、其他注意的写法
         *      1）如果上游是Kafka，直接在 官方提供的SourceFunction实现类上，指定watermark
         *      2）多分区的数据源，设置一个 idle超时，防止 watermark不更新的问题
         *
         * TODO 8、问题：读文件
         *      退出之前，会把 watermark更新为最大值，也就是 Long的最大值
         */
    }
}
